package engine

import "log"

type ConcurrentEngine struct {
	Scheduler        Scheduler
	WorkerCount      int
	ItemChan         chan Item
	RequestProcessor Processor
}

type Processor func(Request) (ParseResult, error)

type Scheduler interface {
	ReadyNotifier
	Submit(Request)
	WorkerChan() chan Request
	Run()
}

type ReadyNotifier interface {
	WorkerReady(chan Request)
}

func (c *ConcurrentEngine) Run(seeds ...Request) {
	out := make(chan ParseResult)
	c.Scheduler.Run()

	//为了不让chanel死锁，把goroutine的方法放在前面等待接受
	for i := 0; i < c.WorkerCount; i++ {
		c.createWorker(c.Scheduler.WorkerChan(), out, c.Scheduler)
	}

	for _, r := range seeds {
		c.Scheduler.Submit(r)
	}

	for {
		result := <-out
		for _, item := range result.Items {
			//获取到Item,然后抛出来，让其他线程存储
			go func() {
				c.ItemChan <- item
			}()
		}

		for _, request := range result.Request {
			if !isDuplicate(request.Url) {
				continue
			}
			c.Scheduler.Submit(request)
		}
	}
}

func (c *ConcurrentEngine) createWorker(in chan Request, out chan ParseResult, ready ReadyNotifier) {
	go func() {
		for {
			//tel scheduler i'm ready
			//in是个chanel，他会因为Scheduler的值的改变而变化
			ready.WorkerReady(in)
			request := <-in
			result, err := c.RequestProcessor(request)
			if err != nil {
				log.Println(err)
				continue
			}
			out <- result
		}
	}()
}
